package com.uuuo.rxjava2;


import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

/**
 * Created by JiaTao on 2018/8/31.
 * Email: 535318720@qq.com
 * Version: 1.0
 * Description:
 */
public class MyMapObservable extends Observable {
    //用来保存事件发布者，也就是被观察者
    private ObservableSource source;

    public MyMapObservable(ObservableSource source) {
        this.source = source;
    }

    //在外部调用了subscribe方法后
    //就会回调到该方法，并且把外部创建的Observer对象传进来
    @Override
    protected void subscribeActual(Observer observer) {
        //创建MyMapObserver对象保存对外部Observer对象的引用
        //调用 Observable.just("1")--->ObservableJust的subscribe方法
        //回调ObservableJust的subscribeActual
        //subscribeActual方法会调用MyMapObserver的onNext
        //MyMapObserver的onNext调用observer的onNext
        source.subscribe(new MyMapObserver(observer));
    }

    class MyMapObserver implements Observer {
        Observer observer;

        public MyMapObserver(Observer observer) {
            this.observer = observer;
        }

        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(Object o) {
            System.out.println("MyMapObserver--->onNext");
            observer.onNext(o);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    }
}
